package com.whfc.emp.mqtt;

import com.whfc.common.face.szyc.SearchPerson;
import com.whfc.common.face.szyc.SzycConst;
import com.whfc.common.mqtt.MqttProperties;
import com.whfc.common.spring.AppContextUtil;
import com.whfc.emp.dao.AppFaceGateMapper;
import com.whfc.emp.entity.AppFaceGate;
import com.whfc.emp.enums.FaceGateType;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.handler.annotation.Header;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;

/**
 * MQTT 配置
 *
 * @author qzexing
 * @version 1.0
 * @date 2021/12/9 15:06
 */
@Configuration
@IntegrationComponentScan
@ConditionalOnProperty(value = "mqtt.enabled", havingValue = "true")
public class MqttConfig {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public static Thread MQTT_THREAD = null;

    public static SearchPerson SEARCH_PERSON = null;

    @Autowired
    private AppFaceGateMapper appFaceGateMapper;

    @Autowired
    private MqttProperties mqttProperties;

    /**
     * MQTT订阅通道
     */
    public static final String MQTT_CHANNEL_IN = "mqttInputChannel";

    /**
     * MQTT发布通道
     */
    public static final String MQTT_CHANNEL_OUT = "mqttOutputChannel";

    /**
     * MQTT客户端
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        logger.info("创建MQTT工厂");
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置连接的用户名
        options.setUserName(mqttProperties.getUsername());
        // 设置连接的密码
        options.setPassword(mqttProperties.getPassword().toCharArray());
        // 设置连接的地址
        options.setServerURIs(new String[]{mqttProperties.getUrl()});
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线
        options.setKeepAliveInterval(20);
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * MQTT信息通道（发布者）
     */
    @Bean(name = MqttConfig.MQTT_CHANNEL_OUT)
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    /**
     * MQTT消息处理器（生产者）
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConfig.MQTT_CHANNEL_OUT)
    public MessageHandler mqttOutbound() {
        //随机 clientId
        Random random = new Random();
        int i = random.nextInt(100);
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                mqttProperties.getClientId() + "_producer" + i,
                mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(0);
        return messageHandler;
    }

    /**
     * MQTT信息通道（消费者）
     */
    @Bean(name = MqttConfig.MQTT_CHANNEL_IN)
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }


    /**
     * MQTT消息订阅绑定（消费者）
     */
    @Primary
    @Bean
    public MessageProducer inbound() {
        // 可以同时消费（订阅）多个Topic
        List<String> deviceKeyList = getFaceGateDeviceKeyList();
        List<String> subscribeTopicList = getSubscribeTopicList(deviceKeyList);
        //随机 clientId
        Random random = new Random();
        int i = random.nextInt(100);
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getClientId() + "_consumer" + i, mqttClientFactory(),
                        subscribeTopicList.toArray(new String[]{}));
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(0);
        // 设置订阅通道
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * MQTT消息处理器（消费者）
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConfig.MQTT_CHANNEL_IN)
    public MessageHandler handler() {
        return message -> {
            String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
            String msg = message.getPayload().toString();
            logger.info("MQTT接收到消息, topic：{}", topic);
            //发布事件
            AppContextUtil.context().publishEvent(new MqttReceiverEvent(msg));
        };
    }

    /**
     * 获取订阅的消息主题列表
     *
     * @param machSnList 设备SN列表
     * @return 的消息主题列表
     */
    private List<String> getSubscribeTopicList(List<String> machSnList) {
        List<String> list = new ArrayList<>();
        for (String machSn : machSnList) {
            // 人员识别
            String rec = MessageFormat.format(SzycConst.TOPIC_REC, machSn);
            list.add(rec);
            // 回复平台
            String ack = MessageFormat.format(SzycConst.TOPIC_ACK, machSn);
            list.add(ack);
            // 陌生人抓拍记录
            String snap = MessageFormat.format(SzycConst.TOPIC_SNAP, machSn);
            list.add(snap);
            // 二维码扫码信息
            String qrCode = MessageFormat.format(SzycConst.TOPIC_QRCODE, machSn);
            list.add(qrCode);
        }
        return list;
    }

    /**
     * 获取所有玉川 MQTT设备序列号
     *
     * @return 玉川 MQTT设备序列号
     */
    private List<String> getFaceGateDeviceKeyList() {
        List<AppFaceGate> list = appFaceGateMapper.selectByPlatform(FaceGateType.MQTT_HQSX_ZJJ.getCode());
        List<AppFaceGate> list1 = appFaceGateMapper.selectByPlatform(FaceGateType.MQTT_HQSX.getCode());
        list.addAll(list1);
        return list.stream().map(AppFaceGate::getDeviceKey).collect(Collectors.toList());
    }

    /**
     * 消息发送中
     */
    @MessagingGateway(defaultRequestChannel = MqttConfig.MQTT_CHANNEL_OUT)
    public interface MqttMessageSender {

        /**
         * 指定主题发送消息
         *
         * @param topic   主题
         * @param payload 消息主体
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);

        /**
         * 指定主题、qos发送消息
         *
         * @param topic   主题
         * @param qos     对消息处理的几种机制。
         *                0 表示的是订阅者没收到消息不会再次发送，消息会丢失。
         *                1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。
         *                2 多了一次去重的动作，确保订阅者收到的消息有一次。
         * @param payload 消息主体
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);

    }
}
